package com.rrd.hive.udf;

import com.weicai.caesar.CaesarUtil;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.ql.exec.Description;
import org.apache.hadoop.hive.ql.exec.UDF;
import org.apache.hadoop.io.Text;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;


@Description(
        name = "mobile_area",
        value = "_FUNC_(mobile) - Get area string from mobile or caesar encode mobile",
        extended = "Example:\n> SELECT _FUNC_('13811360000');\n  北京,北京,移动"
)
public class MobileAreaUDF extends UDF {
    public static final Map<String, String> areaMap = new HashMap<>();
    public static String dataFile = "/data/dim/mobile_area/phone_area";

    static {
        try {
            readAreaFile();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    public static Configuration config() {
        System.setProperty("HADOOP_USER_NAME", "hdfs");
        Configuration conf = new Configuration();
        conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
        return conf;
    }

    public static void readAreaFile() throws IOException {
        Configuration conf = config();
        FileSystem fs = FileSystem.newInstance(URI.create(dataFile), conf);
        Path remotePath = new Path(dataFile);
        FSDataInputStream in = fs.open(remotePath);
        if (fs.exists(remotePath)) {
            BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(in));
            String lineTxt;
            String[] fieldValue;
            while ((lineTxt = bufferedReader.readLine()) != null) {
                fieldValue = lineTxt.split(",");
                areaMap.put(fieldValue[0], lineTxt.substring(fieldValue[0].length() + 1));
            }
            bufferedReader.close();
            in.close();
            fs.close();
        }
        else {
            System.out.println("找不到指定的文件");
        }
    }

    public Text evaluate(String mobile){
        if (mobile == null || mobile.equals("")) {
            return null;
        }
        Text text = new Text();
        if (CaesarUtil.isEncrypted(mobile)) {
            mobile = CaesarUtil.decode(mobile);
        }
        if (mobile.length() != 11) {
            return null;
        }
        String area = areaMap.get(mobile.substring(0, 7));
        if (area != null) {
            text.set(area);
        }
        return text;
    }

}
